﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using UtilZ.Dotnet.Ex.Log;
using UtilZ.Dotnet.Ex.ZLog;

namespace UtilZ.Dotnet.Ex.Base
{
    /// <summary>
    /// 线程池扩展类k
    /// </summary>
    public class ThreadPoolEx : IDisposableZ
    {
        private readonly ThreadPoolThread[] _threads;
        private readonly int _threadCount;
        private readonly ConcurrentQueue<ThreadPoolThread> _pools = new ConcurrentQueue<ThreadPoolThread>();
        private readonly AutoResetEvent _eventHandler = new AutoResetEvent(false);
        private IThreadPoolGroup _threadPoolGroup;


        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="cpuCoreIndexs">指定运行的CPU核心索引,从0开始[0-N]</param>
        public ThreadPoolEx(IEnumerable<int> cpuCoreIndexs)
        {
            if (cpuCoreIndexs == null)
            {
                throw new ArgumentNullException(nameof(cpuCoreIndexs));
            }

            int count = cpuCoreIndexs.Count();
            if (count <= 0 || count > Environment.ProcessorCount)
            {
                throw new ArgumentException($"当前硬件平台总共包含CPU核心数:{Environment.ProcessorCount},指定的CPU核心数{count}超出范围", nameof(cpuCoreIndexs));
            }

            this._threads = new ThreadPoolThread[count];
            int coreIndex;
            for (int i = 0; i < count; i++)
            {
                coreIndex = cpuCoreIndexs.ElementAt(i);

                if (coreIndex < 0 || coreIndex > Environment.ProcessorCount)
                {
                    throw new ArgumentException($"当前硬件平台总共包含CPU核心数:{Environment.ProcessorCount},指定的CPU核心索引\"{coreIndex}\"超出范围", nameof(cpuCoreIndexs));
                }

                this._threads[i] = new ThreadPoolThread(coreIndex);
                this._pools.Enqueue(this._threads[i]);
            }
            this._threadCount = count;
        }


        private List<ThreadPoolThread> ApplyThread(int count)
        {
            var list = new List<ThreadPoolThread>();
            ThreadPoolThread thread;

            while (!this._disposed)
            {
                while (!this._disposed && this._pools.TryDequeue(out thread))
                {
                    list.Add(thread);
                    if (count < this._threadCount && list.Count >= count)
                    {
                        break;
                    }
                }

                if (this._disposed || list.Count > 0)
                {
                    break;
                }

                try
                {
                    this._eventHandler.WaitOne(10);
                }
                catch (ObjectDisposedException)
                {

                }
            }

            return list;
        }

        private void ReleaeThread(List<ThreadPoolThread> list)
        {
            foreach (var thread in list)
            {
                thread.Reset();
                this._pools.Enqueue(thread);
            }

            try
            {
                this._eventHandler.Set();
            }
            catch (ObjectDisposedException)
            {

            }
        }



        /// <summary>
        /// 执行For并行操作
        /// </summary>
        /// <param name="fromInclusive">起始索引</param>
        /// <param name="toExclusive">结束索引</param>
        /// <param name="body">执行回调</param>
        /// <returns>单次执行结果</returns>
        /// <exception cref="ObjectDisposedException">当前对象释放异常</exception>
        public ThreadPoolExcuteResult For(int fromInclusive, int toExclusive, Action<int> body)
        {
            if (body == null)
            {
                throw new ArgumentNullException(nameof(body));
            }

            if (this._disposed)
            {
                throw new ObjectDisposedException(nameof(ThreadPoolEx));
            }

            int count = toExclusive - fromInclusive;
            if (count <= 0)
            {
                throw new ArgumentOutOfRangeException($"没有可迭代的范围,起始结束索引\"{fromInclusive}-{toExclusive}\"无效.", nameof(body));
            }

            List<ThreadPoolThread> list = this.ApplyThread(count);
            if (this._disposed)
            {
                throw new ObjectDisposedException(nameof(ThreadPoolEx));
            }

            using var group = new ThreadPoolForGroup(this, list, fromInclusive, toExclusive, body);
            this._threadPoolGroup = group;
            group.Wait();
            this._threadPoolGroup = null;
            return group.IsCancellationRequested ? ThreadPoolExcuteResult.Cancelled : ThreadPoolExcuteResult.Sucess; ;
        }

        /// <summary>
        /// 执行ForEach并行操作[返回值为true表示取消;falase:表示完成]
        /// </summary>
        /// <typeparam name="TSource">元素类型</typeparam>
        /// <param name="source">用于并行操作的集合</param>
        /// <param name="body">执行回调</param>
        /// <returns>单次执行结果</returns>
        /// <exception cref="ObjectDisposedException">当前对象释放异常</exception>
        public ThreadPoolExcuteResult ForEach<TSource>(IEnumerable<TSource> source, Action<TSource> body)
        {
            if (source == null)
            {
                throw new ArgumentNullException(nameof(source));
            }

            if (body == null)
            {
                throw new ArgumentNullException(nameof(body));
            }

            if (this._disposed)
            {
                throw new ObjectDisposedException(nameof(ThreadPoolEx));
            }

            int count = source.Count();
            if (count <= 0)
            {
                return ThreadPoolExcuteResult.Failed;
            }

            List<ThreadPoolThread> list = this.ApplyThread(count);
            if (this._disposed)
            {
                throw new ObjectDisposedException(nameof(ThreadPoolEx));
            }

            using var group = new ThreadPoolForeachGroup<TSource>(this, list, source, body);
            this._threadPoolGroup = group;
            group.Wait();
            this._threadPoolGroup = null;
            return group.IsCancellationRequested ? ThreadPoolExcuteResult.Cancelled : ThreadPoolExcuteResult.Sucess;
        }


        /// <summary>
        /// 取消执行
        /// </summary>
        public void Cancell()
        {
            IThreadPoolGroup group = this._threadPoolGroup;
            if (group != null)
            {
                group.Cancell();
            }
        }


        private bool _disposed = false;
        /// <summary>
        /// 当前对象释放标识
        /// </summary>
        public bool Disposed
        {
            get { return this._disposed; }
        }

        /// <summary>
        /// 释放资源
        /// </summary>
        public void Dispose()
        {
            if (this._disposed)
            {
                return;
            }
            this._disposed = true;


            foreach (var thread in this._threads)
            {
                thread.Dispose();
            }
            this._eventHandler.Set();
            this._eventHandler.Dispose();
        }


        private class ThreadPoolThread : IDisposable
        {
            private readonly int _coreIndex;
            private readonly Thread _thread;
            private bool _runing = true;
            private readonly AutoResetEvent _disposed = new AutoResetEvent(false);

            private ExccuteTypes _type = ExccuteTypes.None;
            private IThreadPoolGroup _forGroup = null;
            private Action<int> _forBody;
            private Action<object> _foreachBody;

            private readonly AutoResetEvent _eventHandler = new AutoResetEvent(false);
            public ThreadPoolThread(int coreIndex)
            {
                this._coreIndex = coreIndex;

                this._thread = new Thread(new ParameterizedThreadStart(this.ThreadExcuteMethod));
                this._thread.Name = "ThreadPoolEx";
                this._thread.IsBackground = true;
                this._thread.Start();
            }

            /// <summary>
            /// 线程执行方法
            /// </summary>
            /// <param name="obj">线程参数</param>
            private void ThreadExcuteMethod(object obj)
            {
                try
                {
                    ThreadEx.SetThreadRunCPUCore(this._coreIndex);
                    IThreadPoolGroup forGroup;
                    int index;
                    object item;

                    while (this._runing)
                    {
                        forGroup = this._forGroup;
                        if (forGroup != null)
                        {
                            switch (this._type)
                            {
                                case ExccuteTypes.For:
                                    while (forGroup.TryGetIndex(out index))
                                    {
                                        this._forBody?.Invoke(index);
                                    }
                                    break;
                                case ExccuteTypes.Foreach:
                                    while (forGroup.TryGetItem(out item))
                                    {
                                        this._foreachBody?.Invoke(item);
                                    }
                                    break;
                                case ExccuteTypes.None:
                                default:
                                    break;
                            }
                        }

                        try
                        {
                            this._eventHandler.WaitOne(10000);
                        }
                        catch (ObjectDisposedException)
                        {
                        }
                    }
                }
                catch (ThreadAbortException taex)
                {
                    ZLoger.Warn(taex, "内部线程池执行终止");
                }
                catch (Exception ex)
                {
                    ZLoger.Error(ex, "内部线程池执行异常");
                }
            }

            public void SetFor(IThreadPoolGroup group, Action<int> body)
            {
                this._forGroup = group;
                this._forBody = body;
                this._type = ExccuteTypes.For;
                this._eventHandler.Set();
            }

            public void SetForeach(IThreadPoolGroup group, Action<object> body)
            {
                this._forGroup = group;
                this._foreachBody = body;
                this._type = ExccuteTypes.Foreach;

                try
                {
                    this._eventHandler.Set();
                }
                catch (ObjectDisposedException)
                {

                }
            }


            public void Reset()
            {
                this._type = ExccuteTypes.None;
                this._forGroup = null;
                this._forBody = null;
            }


            /// <summary>
            /// 释放资源
            /// </summary>
            public void Dispose()
            {
                this._runing = false;
                this._eventHandler.Dispose();
            }

        }

        private interface IThreadPoolGroup : IDisposable
        {
            bool TryGetIndex(out int index);

            bool TryGetItem(out object index);

            void Wait();

            void Cancell();

            bool IsCancellationRequested { get; }
        }

        private enum ExccuteTypes
        {
            None,

            For,

            Foreach,
        }


        private class ThreadPoolForGroup : IThreadPoolGroup
        {
            private readonly ThreadPoolEx _threadPool;
            private readonly List<ThreadPoolThread> _list;
            private readonly int _toExclusive;
            private readonly AutoResetEvent _handler = new AutoResetEvent(false);
            public ThreadPoolForGroup(ThreadPoolEx threadPool, List<ThreadPoolThread> list, int fromInclusive, int toExclusive, Action<int> body)
            {
                this._threadPool = threadPool;
                this._list = list;
                this._toExclusive = toExclusive;
                this._index = fromInclusive - 1;

                foreach (var thread in this._list)
                {
                    thread.SetFor(this, body);
                }
            }


            private int _index;
            private int _endThreadCount = 0;
            public bool TryGetIndex(out int index)
            {
                index = Interlocked.Increment(ref this._index);
                if (index < this._toExclusive)
                {
                    return true;
                }
                else
                {
                    try
                    {
                        var endThreadCount = Interlocked.Increment(ref this._endThreadCount);
                        if (endThreadCount >= this._list.Count)
                        {
                            this._handler.Set();
                        }
                    }
                    catch (ObjectDisposedException)
                    {

                    }
                    return false;
                }
            }

            bool IThreadPoolGroup.TryGetItem(out object index)
            {
                throw new NotImplementedException();
            }

            public void Wait()
            {
                try
                {
                    this._handler.WaitOne();
                }
                catch (ObjectDisposedException)
                {

                }
            }

            /// <summary>
            /// 取消执行
            /// </summary>
            public void Cancell()
            {
                Interlocked.Exchange(ref this._index, this._toExclusive);
                this._isCancellationRequested = true;
            }

            private bool _isCancellationRequested = false;
            /// <summary>
            /// 取消标识
            /// </summary>
            public bool IsCancellationRequested
            {
                get { return this._isCancellationRequested; }
            }

            public void Dispose()
            {
                this._threadPool.ReleaeThread(this._list);
                this._handler.Dispose();
            }
        }

        private class ThreadPoolForeachGroup<TSource> : IThreadPoolGroup
        {
            private readonly ThreadPoolEx _threadPool;
            private readonly List<ThreadPoolThread> _list;
            private readonly IEnumerable<TSource> _items;
            private readonly int _itemCount;
            private readonly AutoResetEvent _handler = new AutoResetEvent(false);

            public ThreadPoolForeachGroup(ThreadPoolEx threadPool, List<ThreadPoolThread> list, IEnumerable<TSource> items, Action<TSource> body)
            {
                this._threadPool = threadPool;
                this._list = list;
                this._items = items;
                this._itemCount = items.Count();

                foreach (var thread in this._list)
                {
                    thread.SetForeach(this, (obj) =>
                    {
                        body((TSource)obj);
                    });
                }
            }


            bool IThreadPoolGroup.TryGetIndex(out int index)
            {
                throw new NotImplementedException();
            }

            private int _index = -1;
            private int _endThreadCount = 0;
            public bool TryGetItem(out object item)
            {
                int index = Interlocked.Increment(ref this._index);
                if (index < this._itemCount)
                {
                    item = this._items.ElementAt(index);
                    return true;
                }
                else
                {
                    item = null;
                    try
                    {
                        var endThreadCount = Interlocked.Increment(ref this._endThreadCount);
                        if (endThreadCount >= this._list.Count)
                        {
                            this._handler.Set();
                        }
                    }
                    catch (ObjectDisposedException)
                    {

                    }

                    return false;
                }
            }

            public void Wait()
            {
                this._handler.WaitOne();
            }

            /// <summary>
            /// 取消执行
            /// </summary>
            public void Cancell()
            {
                Interlocked.Exchange(ref this._index, this._itemCount);
                this._isCancellationRequested = true;
            }

            private bool _isCancellationRequested = false;
            /// <summary>
            /// 取消标识
            /// </summary>
            public bool IsCancellationRequested
            {
                get { return this._isCancellationRequested; }
            }

            public void Dispose()
            {
                this._threadPool.ReleaeThread(this._list);
                this._handler.Dispose();
            }
        }
    }

    /// <summary>
    /// 线程池执行结果
    /// </summary>
    public enum ThreadPoolExcuteResult
    {
        /// <summary>
        /// 成功
        /// </summary>
        Sucess = 1,

        /// <summary>
        /// 失败
        /// </summary>
        Failed = 2,

        /// <summary>
        /// 取消执行
        /// </summary>
        Cancelled = 3,

        /// <summary>
        /// 当前对象已释放
        /// </summary>
        Disposed = 4,
    }
}
